package com.skivingcloud.report.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mybatisflex.spring.service.impl.ServiceImpl;
import com.skivingcloud.common.exceptions.BusinessException;
import com.skivingcloud.common.utils.JacksonUtil;
import com.skivingcloud.report.constants.JdbcConstants;
import com.skivingcloud.report.dto.DataSetDto;
import com.skivingcloud.report.dto.DataSourceDto;
import com.skivingcloud.report.entity.ReportDataSource;
import com.skivingcloud.report.mapper.ReportDataSourceMapper;
import com.skivingcloud.report.service.JdbcService;
import com.skivingcloud.report.service.ReportDataSourceService;
import com.skivingcloud.report.utils.AnalysisUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import com.mybatisflex.core.paginate.Page;
import com.mybatisflex.core.query.QueryWrapper;
import com.skivingcloud.common.utils.PageParameter;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;

/**
 * 数据源管理 服务层实现。
 *
 * @author skivingcloud
 * @since 2024-09-12
 */
@Service
public class ReportDataSourceServiceImpl extends ServiceImpl<ReportDataSourceMapper, ReportDataSource> implements ReportDataSourceService {
    private static final Logger log = LoggerFactory.getLogger(ReportDataSourceServiceImpl.class);
    private RestTemplate restTemplateOuter;
    private JdbcService jdbcService;

    public ReportDataSourceServiceImpl(@Qualifier("restTemplateOuter") RestTemplate restTemplateOuter, JdbcService jdbcService) {
        this.restTemplateOuter = restTemplateOuter;
        this.jdbcService = jdbcService;
    }
    /**
     * 分页查询数据源管理
     *
     * @param parameter 查询参数
     * @return 分页查询结果
     */
    @Override
    public Page<ReportDataSource> queryByPage(PageParameter<ReportDataSource> parameter){
        Page<ReportDataSource> page = new Page<>(parameter.getPage(), parameter.getPageSize());
        QueryWrapper queryWrapper = QueryWrapper.create();
        if (parameter.getParam() != null) {
            //查询条件
            if (StringUtils.isNoneBlank(parameter.getParam().getId())) {
                queryWrapper.and(ReportDataSource::getId).like(parameter.getParam().getId());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getSourceCode())) {
                queryWrapper.and(ReportDataSource::getSourceCode).like(parameter.getParam().getSourceCode());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getSourceName())) {
                queryWrapper.and(ReportDataSource::getSourceName).like(parameter.getParam().getSourceName());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getSourceDesc())) {
                queryWrapper.and(ReportDataSource::getSourceDesc).like(parameter.getParam().getSourceDesc());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getSourceType())) {
                queryWrapper.and(ReportDataSource::getSourceType).like(parameter.getParam().getSourceType());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getSourceConfig())) {
                queryWrapper.and(ReportDataSource::getSourceConfig).like(parameter.getParam().getSourceConfig());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getEnableFlag())) {
                queryWrapper.and(ReportDataSource::getEnableFlag).like(parameter.getParam().getEnableFlag());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getDeleteFlag())) {
                queryWrapper.and(ReportDataSource::getDeleteFlag).like(parameter.getParam().getDeleteFlag());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getCreateBy())) {
                queryWrapper.and(ReportDataSource::getCreateBy).like(parameter.getParam().getCreateBy());
            }
            if (StringUtils.isNoneBlank(parameter.getParam().getUpdateBy())) {
                queryWrapper.and(ReportDataSource::getUpdateBy).like(parameter.getParam().getUpdateBy());
            }
        }
        return this.page(page, queryWrapper);
    }

    /**
     * 测试数据源连接
     * @param reportDataSource 数据源信息
     * @return 结果
     */
    @Override
    public Boolean testConnect(ReportDataSource reportDataSource) throws JsonProcessingException {
        String sourceType = reportDataSource.getSourceType();
        String sourceConfig = reportDataSource.getSourceConfig();
        DataSourceDto dto = new DataSourceDto();
        dto.setSourceConfig(sourceConfig);
        switch (sourceType) {
            case JdbcConstants.ELASTIC_SEARCH_SQL -> testElasticsearchSqlConnection(dto);
            case JdbcConstants.MYSQL,
                 JdbcConstants.KUDU_IMAPLA,
                 JdbcConstants.ORACLE,
                 JdbcConstants.SQL_SERVER,
                 JdbcConstants.JDBC,
                 JdbcConstants.POSTGRESQL,
                 JdbcConstants.DAMENG,
                 JdbcConstants.OPENGAUSS,
                 JdbcConstants.KINGBASE
                -> testRelationalDb(dto);
            case JdbcConstants.HTTP ->testHttp(dto);
        }
        return true;
    }

    /**
     * 执行sql,统计数据total
     *
     * @param dataSourceDto 数据源
     * @param param         参数
     * @return 数量
     */
    @Override
    public long total(DataSourceDto dataSourceDto, DataSetDto param) throws JsonProcessingException {
        //区分数据类型
        String sourceType = dataSourceDto.getSourceType();
        return switch (sourceType) {
            case JdbcConstants.ELASTIC_SEARCH_SQL -> 0;
            case JdbcConstants.MYSQL -> mysqlTotal(dataSourceDto, param);
            default -> throw new BusinessException("不支持的数据源");
        };
    }

    /**
     * 执行sql
     *
     * @param dataSourceDto 数据源
     * @return 结果
     */
    @Override
    public List<Map<String, Object>> execute(DataSourceDto dataSourceDto) throws JsonProcessingException {
        String sourceType = dataSourceDto.getSourceType();
        return switch (sourceType) {
            case JdbcConstants.ELASTIC_SEARCH_SQL -> executeElasticsearchSql(dataSourceDto);
            case JdbcConstants.MYSQL, JdbcConstants.KUDU_IMAPLA, JdbcConstants.ORACLE, JdbcConstants.SQL_SERVER,
                 JdbcConstants.JDBC, JdbcConstants.POSTGRESQL, JdbcConstants.DAMENG, JdbcConstants.OPENGAUSS,
                 JdbcConstants.KINGBASE -> executeRelationalDb(dataSourceDto);
            case JdbcConstants.HTTP -> executeHttp(dataSourceDto);
            default -> throw new BusinessException("不支持的数据源类型");
        };
    }

    /**
     * Elasticsearch测试
     * @param dto 参数
     */
    private void testElasticsearchSqlConnection(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisHttpConfig(dto);
        String apiUrl = dto.getApiUrl();
        WebClient webClient = WebClient.builder().baseUrl(apiUrl).build();
        Mono<String> result = webClient.get()
                .headers(headers -> {
                    headers.setContentType(MediaType.APPLICATION_JSON);
                    if (StringUtils.isNoneBlank(dto.getUsername())) {
                        headers.setBasicAuth(dto.getUsername(), dto.getPassword());
                    }
                }).retrieve().bodyToMono(String.class);
        result.subscribe(re -> log.info("测试Elasticsearch成功：{}",re), error -> log.error(error.getMessage(), error));
    }

    /**
     * 关系型数据库测试
     * @param dto 参数
     */
    private void testRelationalDb(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisRelationalDbConfig(dto);
        try {
            Connection unPooledConnection = jdbcService.getUnPooledConnection(dto);
            String catalog = unPooledConnection.getCatalog();
            log.info("数据库测试连接成功：{}", catalog);
            unPooledConnection.close();
        } catch (SQLException e) {
            log.error("error",e);
            if (e.getCause() instanceof ClassNotFoundException) {
                throw new BusinessException(e.getCause().getMessage());
            } else {
                throw new BusinessException(e.getMessage());
            }
        }
    }

    /**
     * http接口测试
     * @param dto 参数
     */
    private void testHttp(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisHttpConfig(dto);
        String apiUrl = dto.getApiUrl();
        String method = dto.getMethod();
        String body = dto.getBody();
        HttpHeaders headers = new HttpHeaders();
        headers.setAll(JacksonUtil.jsonStringToMap(dto.getHeader()));
        HttpEntity<String> entity = new HttpEntity<>(body, headers);
        ResponseEntity<Object> exchange;
        try {
            exchange = restTemplateOuter.exchange(apiUrl, HttpMethod.valueOf(method), entity, Object.class);
            if (exchange.getStatusCode().isError()) {
                throw new BusinessException(Objects.requireNonNull(exchange.getBody()).toString());
            }
        } catch (RestClientException e) {
            throw new BusinessException(e.getMessage());
        }
    }

    /**
     * 获取mysql count 和添加limit分页信息
     * @param sourceDto 数据源
     * @param dto 参数
     * @return 数据量
     */
    private long mysqlTotal(DataSourceDto sourceDto, DataSetDto dto) throws JsonProcessingException {
        String dynSentence = sourceDto.getDynSentence();
        String sql = "select count(1) as count from (" + dynSentence + ") as gaeaExecute";
        sourceDto.setDynSentence(sql);
        List<Map<String, Object>> result = execute(sourceDto);

        //sql 拼接 limit 分页信息
        int pageNumber = Integer.parseInt(dto.getContextData().getOrDefault("pageNumber", "1").toString());
        int pageSize = Integer.parseInt(dto.getContextData().getOrDefault("pageSize", "10").toString());
        String sqlLimit = " limit " + (pageNumber - 1) * pageSize + "," + pageSize;
        sourceDto.setDynSentence(dynSentence.concat(sqlLimit));
        log.info("当前total：{}, 添加分页参数,sql语句：{}", JacksonUtil.mapToJsonString(result.get(0)), sourceDto.getDynSentence());
        return (long) result.get(0).get("count");
    }

    /**
     *执行es查询
     * @param dto 参数
     * @return 返回值
     */
    private List<Map<String, Object>> executeElasticsearchSql(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisHttpConfig(dto);
        HttpHeaders headers = new HttpHeaders();
        headers.setAll(JacksonUtil.jsonStringToMap(dto.getHeader()));
        HttpEntity<String> entity = new HttpEntity<>(dto.getDynSentence(), headers);
        ResponseEntity<JsonNode> exchange;
        try {
            exchange = restTemplateOuter.exchange(dto.getApiUrl(), HttpMethod.valueOf(dto.getMethod()), entity, JsonNode.class);
        } catch (Exception e) {
            log.error("error",e);
            throw new BusinessException("数据源连接失败");
        }
        if (exchange.getStatusCode().isError()) {
            throw new BusinessException(JacksonUtil.objectToJsonString(exchange.getBody()));
        }
        List<Map<String, Object>> result;
        try {
            JsonNode body = exchange.getBody();
            //解析es sql数据
            if (null == body) {
                return null;
            }

            List<JsonNode> columns = body.findValues("columns");
            List<JsonNode> rows = body.findValues("rows");
            result = new ArrayList<>();
            for (JsonNode row : rows) {
                Map jsonObject = new HashMap();
                for (int j = 0; j < row.size(); j++) {
                    String name = columns.get(j).findValue("name").textValue();
                    String value = row.get(j).textValue();
                    jsonObject.put(name, value);
                }
                result.add(jsonObject);
            }
        } catch (Exception e) {
            log.error("error",e);
            throw new BusinessException("数据转换失败");
        }
        return result;
    }

    public List<Map<String, Object>> executeRelationalDb(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisRelationalDbConfig(dto);
        Connection pooledConnection = null;
        try {
            pooledConnection = jdbcService.getPooledConnection(dto);

            PreparedStatement statement = pooledConnection.prepareStatement(dto.getDynSentence());
            ResultSet rs = statement.executeQuery();

            int columnCount = rs.getMetaData().getColumnCount();

            List<String> columns = new ArrayList<>();
            for (int i = 1; i <= columnCount; i++) {
                String columnName = rs.getMetaData().getColumnLabel(i);
                columns.add(columnName);
            }
            List<Map<String, Object>> list = new ArrayList<>();
            while (rs.next()) {
                Map jo = new HashMap();
                columns.forEach(t -> {
                    try {
                        Object value = rs.getObject(t);
                        //数据类型转换
                        Object result = dealResult(value);
                        jo.put(t, result);
                    } catch (SQLException throwable) {
                        log.error("error",throwable);
                        throw new BusinessException("执行SQL查询失败");
                    }
                });
                list.add(jo);
            }
            return list;
        } catch (Exception throwable) {
            log.error("error",throwable);
            throw new BusinessException("执行SQL查询失败");
        } finally {
            try {
                if (pooledConnection != null) {
                    pooledConnection.close();
                }
            } catch (SQLException throwable) {
                log.error("error",throwable);
            }
        }
    }

    /**
     * 解决sql返回值 类型问题
     * (through reference chain: java.util.HashMap["pageData"]->java.util.ArrayList[0]->java.util.HashMap["UPDATE_TIME"]->oracle.sql.TIMESTAMP["stream"])
     * @param result 结果
     * @return 转换后结果
     */
    private Object dealResult(Object result) {
        if (null == result) {
            return null;
        }
        String type = result.getClass().getName();
        if ("oracle.sql.TIMESTAMP".equals(type)) {
            //oracle.sql.TIMESTAMP处理逻辑
            return new Date((Long) result);
        }

        return result;
    }

    /**
     * http 执行获取数据
     *
     * @param dto 参数
     */
    public List<Map<String, Object>> executeHttp(DataSourceDto dto) throws JsonProcessingException {
        AnalysisUtil.analysisHttpConfig(dto);
        HttpHeaders headers = new HttpHeaders();
        headers.setAll(JacksonUtil.jsonStringToMap(dto.getHeader()));
        HttpEntity<String> entity = new HttpEntity<>(dto.getDynSentence(), headers);
        ResponseEntity<JsonNode> exchange;
        try {
            exchange = restTemplateOuter.exchange(dto.getApiUrl(), HttpMethod.valueOf(dto.getMethod()), entity, JsonNode.class);
        } catch (Exception e) {
            log.error("error",e);
            throw new BusinessException("数据源连接失败");
        }
        if (exchange.getStatusCode().isError()) {
            throw new BusinessException("数据源连接失败");
        }
        JsonNode body = exchange.getBody();
        String jsonStr = new ObjectMapper().writeValueAsString(body);
        List<Map<String, Object>> result = new ArrayList<>();
        if (jsonStr.trim().startsWith("{") && jsonStr.trim().endsWith("}")) {
            //JSONObject
            result.add(new ObjectMapper().readValue(jsonStr, new TypeReference<>() {
            }));
        } else if (jsonStr.trim().startsWith("[") && jsonStr.trim().endsWith("]")) {
            //List

            result = new ObjectMapper().readValue(jsonStr, new TypeReference<>() {
            });
        } else {
            result.add(new HashMap<>());
        }
        return result;
    }
}
